package com.calabar.phm.etl.driver.spark;

import com.calabar.phm.etl.driver.spark.exception.OperatorParamsValidateException;
import com.calabar.phm.etl.driver.spark.exception.OperatorRunningException;
import com.calabar.phm.etl.operator.api.*;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.annotations.Expose;
import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.Logger;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;

/**
 * @Author zmc <mingcheng.zhang@cdcalabar.com>
 * @Date 17-5-24 下午1:17
 * @Version v1.0
 * @Des 算子的包装类
 */
public class Operator {

    private static final Logger LOG = Logger.getLogger(Operator.class);

    /**
     * 算子唯一标识
     */
    @Expose
    private String operatorUnique;
    /**
     * 算子类型，0：输入算子，1：处理算子，2：输出算子
     */
    @Expose
    private int operatorType;
    /**
     * 算子开发者
     */
    @Expose
    private String operatorDeveloper;
    /**
     * 算子名称
     */
    @Expose
    private String operatorName;
    /**
     * 算子类名
     */
    @Expose
    private String operatorClassName;
    /**
     * 算子参数
     */
    @Expose
    private String operatorParams;
    /**
     * 算子持久化地址
     */
    @Expose
    private String operatorPersist;

    /**
     * Json字符串解析工具
     */
    private Gson gson;

    public Operator() {
        gson = new GsonBuilder().excludeFieldsWithoutExposeAnnotation().setPrettyPrinting().create();
    }


    /**
     * 调用输入算子计算逻辑
     * @param session
     * @return
     * @throws Exception
     */
    public Dataset<Row> transform(SparkSession session) throws
            OperatorParamsValidateException, OperatorRunningException {

        Dataset<Row> transform = null;

        Object instance = getInstance();
        ITransformer transformer = (ITransformer) instance;
        if (this.operatorType != 0) {//不是输入算子
            if (isOpenReadPersist()) {//开启持久化数据读取功能
                //读取持久化数据
                LOG.info("算子（" + this.toString() + ")读取持久化数据，表名称：" + this.operatorPersist);
                try {
                    transform = transformer.readPersistData(session, this.operatorPersist);
                } catch (Exception e) {
                    throw new OperatorRunningException(this.operatorUnique, "", e);
                }
            } else {
                throw new OperatorParamsValidateException(this.operatorUnique, "不是输入算子且没有开启读取缓存功能，无法作为根节点！");
            }
        } else {//输入算子
            InputTransformer inputTransformer = (InputTransformer) instance;
            try {
                transformer.validateParams();
            } catch (Exception ex) {
                throw new OperatorParamsValidateException(this.operatorUnique, ex);
            }
            try {
                transform = inputTransformer.transform(session);
            } catch (Exception ex) {
                throw new OperatorRunningException(this.operatorUnique, ex);
            }
        }

        return transform;
    }

    /**
     *调用计算算子计算逻辑
     * @param session
     * @param dfs
     * @return
     * @throws Exception
     */
    public Dataset<Row> transform(SparkSession session, Dataset<Row>... dfs) throws
            OperatorParamsValidateException, OperatorRunningException {

        Dataset<Row> transform = null;

        Object instance = getInstance();
        ITransformer transformer = (ITransformer) instance;
        ComputeTransformer computeTransformer = (ComputeTransformer) instance;

        try {
            transformer.validateParams();
        } catch (Exception ex) {
            throw new OperatorParamsValidateException(this.operatorUnique, ex);
        }

        try {
            transform = computeTransformer.transform(session, dfs);
        } catch (Exception ex) {
            throw new OperatorRunningException(this.operatorUnique, ex);
        }

        if (isOpenWritePersist()) {//开启持久化数据写入功能
            LOG.info("算子（" + this.toString() + ")写出持久化数据，表名称：" + this.operatorPersist);
            try {
                transformer.writePersistData(transform, this.operatorPersist);
            } catch (Exception ex) {
                throw new OperatorRunningException(this.operatorUnique, ex);
            }
        }


        return transform;
    }

    /**
     * 调用输出算子计算逻辑
     * @param session
     * @param df
     * @throws Exception
     */
    public void transform(SparkSession session, Dataset<Row> df) throws
            OperatorParamsValidateException, OperatorRunningException {

        Object instance = getInstance();
        ITransformer transformer = (ITransformer) instance;
        OutputTransformer outputTransformer = (OutputTransformer) instance;
        try {
            transformer.validateParams();
        } catch (Exception ex) {
            throw new OperatorParamsValidateException(this.operatorUnique, ex);
        }
        try {
            outputTransformer.transform(session, df);
        } catch (Exception ex) {
            throw new OperatorRunningException(this.operatorUnique, ex);
        }
    }

    /**
     * 根据算子类名，反射算子对象
     * @return
     * @throws OperatorParamsValidateException
     */
    public Object getInstance() throws OperatorParamsValidateException {
        Object newInstance = null;
        try {
            Class<?> operatorClass = Class.forName(this.operatorClassName);
            Constructor<?> constructor = operatorClass.getConstructor(String.class, String.class);
            newInstance = constructor.newInstance(this.operatorUnique, this.operatorParams);
        } catch (Throwable e) {
            throw new OperatorParamsValidateException(this.operatorUnique, "根据算子类名反射算子对象出现错误!", e);
        }
        return newInstance;
    }


    @Override
    public String toString() {
        return gson.toJson(this);
    }

    @Override
    public boolean equals(Object o) {
        if (o instanceof Operator) {
            Operator operator = (Operator) o;
            return this.operatorUnique.equals(operator.operatorUnique);
        } else {
            return false;
        }
    }

    @Override
    public int hashCode() {
        return operatorUnique.hashCode();
    }

    /**
     * 根据参数是否为空判断是否开启读取持久化数据的功能
     *
     * @return
     */
    public boolean isOpenReadPersist() {
        return this.operatorPersist != null && !this.operatorPersist.isEmpty();
    }

    /**
     * 根据参数是否为空判断是否开启写入持久化数据的功能
     *
     * @return
     */
    public boolean isOpenWritePersist() {
        return isOpenReadPersist();
    }

    public String getOperatorUnique() {
        return operatorUnique;
    }


    public void setOperatorUnique(String operatorUnique) {
        this.operatorUnique = operatorUnique;
    }

    public void setOperatorId(String operatorId) {
        this.operatorDeveloper = operatorId;
    }

    public String getOperatorName() {
        return operatorName;
    }

    public void setOperatorName(String operatorName) {
        this.operatorName = operatorName;
    }

    public String getOperatorClassName() {
        return operatorClassName;
    }

    public void setOperatorClassName(String operatorClassName) {
        this.operatorClassName = operatorClassName;
    }

    public String getOperatorParams() {

        return this.operatorParams;
    }

    public void setOperatorParams(String operatorParams) {

        this.operatorParams = operatorParams;
    }

    public int getOperatorType() {
        return operatorType;
    }

    public void setOperatorType(int operatorType) {
        this.operatorType = operatorType;
    }

    public String getOperatorDeveloper() {
        return operatorDeveloper;
    }

    public void setOperatorDeveloper(String operatorDeveloper) {
        this.operatorDeveloper = operatorDeveloper;
    }

    public String getOperatorPersist() {
        return operatorPersist;
    }

    public void setOperatorPersist(String operatorPersist) {
        this.operatorPersist = operatorPersist;
    }
}

